package com.heima.kafka;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

public class ProducerTest {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 设置属性
        Properties props = new Properties();
        // 指定连接的kafka服务器的地址
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.137.136:9092");
        // 构建kafka生产者对象
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props, new StringSerializer(), new StringSerializer());
        // 定义kafka主题
        String topic = "sy_160_1";
        // 调用生产者发送消息
        String message = "hello kafka!";
        // 构建消息
        ProducerRecord<String, String> record = new ProducerRecord<String, String>(topic, message);
        // 同时发送key和value
        record = new ProducerRecord<String, String>(topic, "10010", message);
        // 发送消息--同步/异步发送
        Future<RecordMetadata> future = producer.send(record);
        RecordMetadata recordMetadata = future.get();
        System.out.println(recordMetadata);

        // 释放连接
        producer.close();
    }
}
